package com.xujianlong.combat;

import com.sun.xml.internal.bind.v2.schemagen.xmlschema.List;
import com.xujianlong.bean.ApacheLog;
import com.xujianlong.bean.HotUrlCount;
import com.xujianlong.bean.ItemCount;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.Keyed;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;

public class Flink04_Project_HotUrl {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);

        DataStreamSource<String> source = env.readTextFile("input/apache.log");
        WatermarkStrategy<ApacheLog> watermarkStrategy = WatermarkStrategy.<ApacheLog>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<ApacheLog>() {
            @Override
            public long extractTimestamp(ApacheLog apacheLog, long l) {
                return apacheLog.getTs();
            }
        });
        SingleOutputStreamOperator<ApacheLog> apacheLogWithWMDS = source.map(new MapFunction<String, ApacheLog>() {
            @Override
            public ApacheLog map(String s) throws Exception {
                String[] s1 = s.split(" ");
                SimpleDateFormat sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                return new ApacheLog(s1[0],
                        s1[1],
                        sdf.parse(s1[3]).getTime(),
                        s1[5],
                        s1[6]
                );
            }
        }).assignTimestampsAndWatermarks(watermarkStrategy);
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = apacheLogWithWMDS.map(new MapFunction<ApacheLog, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(ApacheLog apacheLog) throws Exception {
                return Tuple2.of(apacheLog.getUrl(), 1);
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyedDS = map.keyBy(data -> data.f0);

        SingleOutputStreamOperator<HotUrlCount> urlCountByWindowDS = keyedDS
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .allowedLateness(Time.seconds(60))
                .sideOutputLateData(new OutputTag<Tuple2<String,Integer>>("sideOutPut"){})
                .aggregate(new HotUrlAggFunc(), new HotUrlWindowFunc());

        KeyedStream<HotUrlCount, Long> windowKeyedDS = urlCountByWindowDS.keyBy(data -> data.getWindowEnd());
        SingleOutputStreamOperator<String> result = windowKeyedDS.process(new HotUrlProcessFunc(5));

        result.print("Result");
        env.execute("ApacheLog");
    }
    public static class HotUrlAggFunc implements AggregateFunction<Tuple2<String,Integer>,Integer,Integer>{

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(Tuple2<String, Integer> stringIntegerTuple2, Integer integer) {
            return integer+1;
        }

        @Override
        public Integer getResult(Integer integer) {
            return integer;
        }

        @Override
        public Integer merge(Integer integer, Integer acc1) {
            return integer+acc1;
        }
    }
    public static class HotUrlWindowFunc implements WindowFunction<Integer, HotUrlCount,String, TimeWindow>{

        @Override
        public void apply(String url, TimeWindow window, Iterable<Integer> input, Collector<HotUrlCount> out) throws Exception {
            Integer count = input.iterator().next();
            out.collect(new HotUrlCount(url,window.getEnd(),count));
        }


    }
    public static class HotUrlProcessFunc extends KeyedProcessFunction<Long,HotUrlCount,String>{

        private Integer totalSize;

        private MapState<String,HotUrlCount> mapState;
        public HotUrlProcessFunc(Integer totalSize){
            this.totalSize=totalSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState=getRuntimeContext().getMapState(new MapStateDescriptor<String, HotUrlCount>("map_state",String.class,HotUrlCount.class));
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            if(timestamp==ctx.getCurrentKey()+60001L){
                mapState.clear();
                return ;
            }

            Iterator<Map.Entry<String, HotUrlCount>> iterator = mapState.iterator();
            ArrayList<Map.Entry<String, HotUrlCount>> entries = Lists.newArrayList(iterator);

            entries.sort((o1,o2)->o2.getValue().getCount()-o1.getValue().getCount());
            StringBuilder sb = new StringBuilder();
            sb.append("========").append(new Timestamp(timestamp-1000L))
                    .append("========")
                    .append("\n");
            for (int i=0;i<Math.min(totalSize,entries.size());i++){
                HotUrlCount itemCount = entries.get(i).getValue();
                sb.append("Top").append(i+1);
                sb.append(" URL:"+itemCount.getUrl());
                sb.append(" Count:").append(itemCount.getCount());
                sb.append("\n");
            }
            sb.append("========").append(new Timestamp(timestamp-1000L))
                    .append("========")
                    .append("\n")
                    .append("\n");


            out.collect(sb.toString());

            Thread.sleep(2000);


        }

        @Override
        public void processElement(HotUrlCount value, Context ctx, Collector<String> out) throws Exception {

            mapState.put(value.getUrl(),value);

            ctx.timerService().registerEventTimeTimer(value.getWindowEnd()+1000L);
            ctx.timerService().registerEventTimeTimer(value.getWindowEnd()+60000L+1L);
            //out.collect("");
        }
    }
}
